package com.atguigu.edu.realtime.util;

import com.atguigu.edu.realtime.common.Constant;

public class SQLUtil {
    public static String getConnectorKafkaSource(String topic, String groupId) {
        return "with(" +
                " 'connector'='kafka', " +
                " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
                " 'properties.group.id'='" + groupId + "', " +
                " 'topic'='" + topic + "', " +
                " 'format'='json', " +
                " 'scan.startup.mode'='latest-offset' " +
                ")";
    }

    public static String getKafkaSink(String topic) {
        return "with(" +
                " 'connector'='kafka', " +
                " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
                " 'topic'='" + topic + "', " +
                " 'format'='json' " +
                ")";
    }

    public static String getUpsertKafkaSink(String topic) {
        return "with(" +
                " 'connector'='upsert-kafka', " +
                " 'properties.bootstrap.servers'='" + Constant.KAFKA_BROKERS + "', " +
                " 'topic'='" + topic + "', " +
                " 'key.format'='json', " +
                " 'value.format'='json' " +
                ")";
    }
}
